package com.hulk.rpc.transport.client.codec;

import com.hulk.rpc.serialize.Serializer;
import com.hulk.rpc.transport.client.future.FutureContainer;
import com.hulk.rpc.transport.client.future.RequestWithFuture;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

import static com.hulk.rpc.config.common.HulkConstants.EXPIRE_PERIOD;

/**
 * @author xuxukang
 */
@Slf4j
public class RequestEncoder extends MessageToByteEncoder<RequestWithFuture> {

  private final Serializer serializer;
  private final FutureContainer futureContainer;

  public RequestEncoder(Serializer serializer, FutureContainer futureContainer) {
    this.serializer = serializer;
    this.futureContainer = futureContainer;
  }

  @Override
  protected void encode(ChannelHandlerContext ctx, RequestWithFuture requestWithFuture, ByteBuf buffer)
          throws Exception {
    futureContainer.add(requestWithFuture);
    serializer.writeRequest(buffer, requestWithFuture.getRequest());

    requestWithFuture.setRequest(null);// help to gc
  }

  @Override
  public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
                      ChannelPromise promise) throws Exception {
    super.connect(ctx, remoteAddress, localAddress, promise);

    if (ctx.channel().attr(CodecConstants.STARTED_AUTO_EXPIRE_JOB).compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
      ctx.executor().scheduleAtFixedRate(//
              () -> futureContainer.doExpireJob(1), //
              EXPIRE_PERIOD, EXPIRE_PERIOD, TimeUnit.MILLISECONDS);

      if (log.isInfoEnabled()) {
        log.info("FutureContainer startingAutoExpireJob");
      }
    }

    if (log.isInfoEnabled()) {
      log.info("channel connect: " + ctx.channel());
    }
  }

  @Override
  public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    super.disconnect(ctx, promise);

    if (log.isInfoEnabled()) {
      log.info("channel disconnect: " + ctx.channel());
    }
  }

  @Override
  public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
    super.close(ctx, promise);

    futureContainer.close();

    if (log.isInfoEnabled()) {
      log.info("channel close: " + ctx.channel());
    }
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

    if (log.isErrorEnabled()) {
      log.error("Exception caught on " + ctx.channel(), cause);
    }

    ctx.channel().close();
    futureContainer.close();
  }
}
